package com.atguigu.Flink.sql.connector;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Flink05_JdbcConnector {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

         //jbcd connector
        String createSourceTable =
                " create table t1(" +
                        " id STRING , " +
                        " vc INT , " +
                        " ts BIGINT " +
                        " ) with (" +
                        "   'connector' = 'jdbc', " +
                        "   'url' = 'jdbc:mysql://hadoop102:3306/test' ," +
                        "   'table-name' = 's1' ," +
                        "   'username' = 'root' ," +
                        "   'password' = '000000'  " +
                        " )" ;

        tableEnv.executeSql(createSourceTable);

        //tableEnv.sqlQuery("select * from t1").execute().print();

        String createSinkTable =
                " create table t2(" +
                        " id STRING , " +
                        " vc INT , " +
                        " ts BIGINT ,  " +
                        " PRIMARY KEY (id) NOT ENFORCED " +
                        " ) with (" +
                        "   'connector' = 'jdbc', " +
                        "   'url' = 'jdbc:mysql://hadoop102:3306/test' ," +
                        "   'table-name' = 's2' ," +
                        "   'username' = 'root' ," +
                        "   'password' = '000000'  " +
                        " )" ;
        tableEnv.executeSql(createSinkTable);

        tableEnv.executeSql("insert into t2 select id,vc,ts from t1");

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
